/**
 *
 * Copyright Pact Lab of H.I.T.
 *
 * Designed and Implemented by Grid Researching Group, 
 * Pact Lab, Harbin
 * 
 * This Project is part of the national 973 Project:
 * Internet Based Virtual Computing Environment
 *
 * http://pact518.hit.edu.cn
 * 
 * Author:       Meteor <meteorlxk@gmail.com> 
 * Copyright:    pact518 
 * Version:      1.0
 * Created:      2009-4-15 
 * LastModified: 2009-4-15
 */
package edu.hit.pact.pgse.master;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.core.session.IoSession;


import vizster.Vizster;
import vizster.VizsterLib;
import edu.hit.pact.data.SpeedData;
import edu.hit.pact.pgse.bean.Task;
import edu.hit.pact.pgse.bean.TaskPiece;
import edu.hit.pact.pgse.bean.TaskPieceStatistics;
import edu.hit.pact.pgse.ch.ConsistentHash;
import edu.hit.pact.pgse.communication.message.AbstractMessage;
import edu.hit.pact.pgse.communication.message.TaskPiecePushMessage;
import edu.hit.pact.pgse.communication.message.TaskPieceStatisticsMessage;
import edu.hit.pact.pgse.graph.GraphLoader;
import edu.hit.pact.pgse.master.util.DistributedMap;
import edu.hit.pact.pgse.util.Globals;
import edu.hit.pact.plot.PlotProcessor;


/**
 * @author Meteor
 * 
 */
public class MasterManager{
	private Log logger;

	private int piecesNumber;
	private DistributedMap<Long, TaskPiece> depository;
	private DistributedMap<Long, InetAddress> pieceCrawlerMap;
	private ConsistentHash<InetAddress> consistentHash;
	private HashMap<InetAddress, IoSession> crawlerSessionMap;
	private IoSession clientSession;
	
	private GraphLoader graphLoader;
	private Vizster vizster;
	private PlotProcessor plotProcessor;
	

	
//	private ConcurrentHashMap<IoSession, SpeedData> speedDataMap;
//	private Timer queueInfoRecorder;
	
	public MasterManager(ConsistentHash<InetAddress> consistentHash) {
		this(LogFactory.getLog(MasterManager.class), consistentHash);
	}

	public MasterManager(Log logger, ConsistentHash<InetAddress> consistentHash) {
		this.logger = logger;
		
		this.piecesNumber = 0;
		this.depository = new DistributedMap<Long, TaskPiece>();
		this.pieceCrawlerMap = new DistributedMap<Long, InetAddress> ();
		this.consistentHash = consistentHash;
		this.crawlerSessionMap = new HashMap<InetAddress, IoSession>();

//		this.queueInfoRecorder = new Timer();
//		queueInfoRecorder.schedule(new DSInfoOutput(), 1*1000, 1*1000);
		
		if (Globals.MASTER_GRAPH_OPEN){
			graphLoader = new GraphLoader();
		    graphLoader.addMaster(Globals.MASTER_ADDRESS_IP);
		    VizsterLib.setLookAndFeel();
		    vizster = new Vizster();
		    vizster.loadGraph(graphLoader.getGraph());
		}
        
		//init plot
		if (Globals.MASTER_PLOT_OPEN){
			plotProcessor = new PlotProcessor();
			plotProcessor.start();
		}
	}

	public synchronized boolean processCrawlerRegister(IoSession session){
		logger.info(session);
		InetAddress ip = ((InetSocketAddress)session.getRemoteAddress()).getAddress();
		crawlerSessionMap.put(ip, session);
		this.consistentHash.add(ip);
		return true;
	}

	public synchronized boolean processClientRegister(IoSession session){
		clientSession = session;
		return true;
	}

	public synchronized void processCrawlerDead(IoSession session) {
		InetAddress ip = ((InetSocketAddress)session.getRemoteAddress()).getAddress();
		logger.error("remove " + ip + " from consistent hash and crawlerSessionMap");
		crawlerSessionMap.put(ip, null);
		consistentHash.remove(ip);
	}
	
	public boolean restartPieces(InetAddress address){
		logger.error("restartPieces");
		
		if (crawlerSessionMap.get(address) != null){//if after waitting is still null, master will restart pieces
			return false;
		}
		
		List<Long> ids = this.pieceCrawlerMap.getIdRemoveElement(address);

		return startTaskPieces(depository.getValues(ids));
	}
	
	public synchronized boolean processFinishedNotify(TaskPieceStatistics statistics) {
		Long taskPieceId = statistics.getTaskPieceId();

		logger.info("remove " + taskPieceId + " from running map and depository");
		this.pieceCrawlerMap.remove(taskPieceId);
		this.depository.remove(taskPieceId);

		logger.info("send statistics to client");
		AbstractMessage message = new TaskPieceStatisticsMessage(statistics);
		if (clientSession != null)
			this.clientSession.write(message);
		else
			logger.error("master can not connect to Client");

		return true;
	}

	public synchronized boolean processTaskPush(Task task) {
		piecesNumber += task.getPieces().size();
		
		Collection<TaskPiece> pieces = task.getPieces();
		logger.info("add task info to taskPieceDepository");
		for (TaskPiece piece: pieces)
			this.depository.add(piece.getId(), piece);
		
		return startTaskPieces(task.getPieces());
	}
	
	private boolean startTaskPieces(Collection<TaskPiece> pieces) {
		logger.info("deal with each task piece.");
		
		for (TaskPiece piece : pieces) {

			logger.info("begin to process " + piece);
			Long taskPieceId = piece.getId();
			String homePage = piece.getHomePage();
			InetAddress node = this.consistentHash.get(homePage);
			
			if (node == null)//no crawler connects with master
				return false;
			
			logger.info("consistent hash return " + node + " to process hostname : " + piece.getId());
			IoSession thisSession = this.crawlerSessionMap.get(node);
			AbstractMessage message = new TaskPiecePushMessage(piece);
			thisSession.write(message);
			this.pieceCrawlerMap.add(taskPieceId, node);
		}
		return true;
	}
	
	public synchronized void processSpeedReport(String crawlerip, String masterip, int speed){
		if (!Globals.MASTER_GRAPH_OPEN)
			return;
		
		graphLoader.updateSpeed(crawlerip, masterip, speed);
	}
	
	public void removePlotNode(String nodeName){
		if (!Globals.MASTER_PLOT_OPEN)
			return;
		
		plotProcessor.removeNode(nodeName);
	}
	
	public void operate(String entity, String operation, IoSession session){
		if (!Globals.MASTER_GRAPH_OPEN)
			return;
		
		if (entity.equals("crawler") && operation.equals("add")){
			graphLoader.addCrawler(session.getRemoteAddress().toString());
		} else if (entity.equals("client") && operation.equals("add")) {
			graphLoader.addClient(session.getRemoteAddress().toString());
		} else if (entity.equals("crawler") && operation.equals("remove")) {
			graphLoader.removeCrawler(session.getRemoteAddress().toString());
		} else if (entity.equals("client") && operation.equals("remove")) {
			graphLoader.removeClient(session.getRemoteAddress().toString());
		} else {
			logger.error("error!");
		}
		vizster.loadGraph(graphLoader.getGraph());
	}
	
//	class DSInfoOutput extends TimerTask{
//		private PrintWriter queueTracker= null;
//		private SimpleDateFormat formater = new SimpleDateFormat("yyyy-MM-dd*hh:mm:ss");
//		public DSInfoOutput(){
//			try {
//				queueTracker = new PrintWriter(new FileOutputStream("MasteranagerInfo.txt", true), true);
//				queueTracker.println("\t\t\ttotal\tdepo\tdistr\tnodetask");
//			} catch (FileNotFoundException e) {
//				// TODO Auto-generated catch block
//				e.printStackTrace();
//			}
//		}
//	
//		public void run() {
//			queueTracker.println(formater.format(new Date())+ "\t" + piecesNumber + "\t" +depository.getPiecesNumber() + "\t" + pieceCrawlerMap.getPiecesNumber());
//		}
//	}
}
